package com.lmh.utils;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lmh.source.JdbcSource;
import org.apache.flume.Event;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.event.EventBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;

/**
 * @program flume-component
 * @Description 关于数据库工具类
 * @Author limeihua
 * @Date 2024/09/02 18:10
 * @Version 1.0
 */
public class DateBaseUtil {
    private static DruidDataSource mysqlDataSource;
    private static final Logger logger = LoggerFactory.getLogger(DateBaseUtil.class);
    public static void paginationExecute(String executeSql, String url, String username, String password, ChannelProcessor channelProcessor, String separator) throws SQLException {
        DruidDataSource DataSource = getMysqlDataSource(url, username, password);
        DruidPooledConnection connection = DataSource.getConnection();
        PreparedStatement statement = connection.prepareStatement(executeSql);
        ResultSet resultSet = statement.executeQuery();
        ResultSetMetaData metaData = resultSet.getMetaData();
        int columnCount = metaData.getColumnCount();
        while (resultSet.next()) {
            String value = "";
            for (int i = 1; i <= columnCount; i++) {
                if(i < columnCount){
                    value = value + resultSet.getObject(i).toString() + separator;
                }else {
                    value = value + resultSet.getObject(i).toString();
                }
            }
            logger.info("采集到的数据为：" + value);
            Event event = EventBuilder.withBody(value.getBytes());
            channelProcessor.processEvent(event);
        }
        if(connection != null){
            connection.close();
        }
    }

    public static DruidDataSource getMysqlDataSource(String url, String username, String password){
        if(mysqlDataSource == null){
            mysqlDataSource = new DruidDataSource();
            mysqlDataSource.setUrl(url);
            mysqlDataSource.setUsername(username);
            mysqlDataSource.setPassword(password);
            mysqlDataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
            mysqlDataSource.setInitialSize(10);
            mysqlDataSource.setMinIdle(10);
            mysqlDataSource.setMaxActive(30);
            mysqlDataSource.setMaxWait(30000);
            mysqlDataSource.setTimeBetweenEvictionRunsMillis(60000);
            mysqlDataSource.setMinEvictableIdleTimeMillis(300000);
            mysqlDataSource.setValidationQuery("SELECT 1");
        }
        return mysqlDataSource;
    }
}
